package com.yy.job.core.run.worker;

import java.security.SecureRandom;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yy.job.common.utils.IPUtils;
import com.yy.job.common.utils.date.YYDateUtils;
import com.yy.job.constant.JobConstant;
import com.yy.job.core.config.xml.ModuleXmlConfig;
import com.yy.job.core.dataprocess.LocalJobDataProcess;
import com.yy.job.core.dataprocess.NewJobDataProcess;
import com.yy.job.core.dataprocess.PendingJobDataProcess;
import com.yy.job.core.dataprocess.PriorityJobDataProcess;
import com.yy.job.core.domain.JobDomain;

/**
 * yy-job的后台线程，一直保持运行状态<br>
 * 每个module对应一个moduleWorker<br>
 * 
 * @author eleven.li
 *
 */
public class ModuleWorker implements Runnable {

	private Logger log = LoggerFactory.getLogger(ModuleWorker.class);

	// 线程执行次数，计数用的
	private int runCount = 0;

	// 默认并发处理数量为1
	public int jobWorkerSize = 1;
	
	// 每次轮询的间隔时间
	public static final int LOOP_SLEEP_SECONDS = 45;

	// 根据wokerCount的数量来创建
	public JobWorker[] jobWorkers = null;

	// 上一次执行时间，用来控制下执行速度
	private long lastExecuteTime = 0L;

	public AtomicInteger wokerIndex = new AtomicInteger(0);

	/**
	 * 判断当前jobBoss是否处于运行状态
	 */
	public boolean isRuning = false;

	/**
	 * moduleWorker名称
	 */
	private String moduleWorkerName;
	
	private String moduleWorkerNameFormat = "%s@%s";
	
	/**
	 * 项目code
	 */
	private String projectCode;
	
	/**
	 * xml配置信息
	 */
	private ModuleXmlConfig moduleJobConfig;

	/**
	 * 高优先级job处理器
	 */
	private LocalJobDataProcess localJobDataProcess;
	
	/**
	 * 高优先级任务处理器，提交的任务在本机立即执行
	 */
	private PriorityJobDataProcess priorityJobDataProcess;

	/**
	 * 状态为(F,L)的job处理器
	 */
	private PendingJobDataProcess pendingJobDateProcess;

	/**
	 * 状态为(C)的job处理器
	 */
	private NewJobDataProcess newJobDataProcess;

	/**
	 * 分布式部署条件下，本机提交的本机执行任务会加入此列表，锁定且在本机执行
	 */
	private Queue<JobDomain> localJobQueue = new ConcurrentLinkedQueue<>();
	
	public ModuleWorker(String projectCode, ModuleXmlConfig moduleJobConfig) {
		this.projectCode = projectCode;
		this.moduleJobConfig = moduleJobConfig;
		moduleJobConfig.setModuleWorker(this);
		
		// 格式化moduleWorker名称
		this.moduleWorkerName = String.format(moduleWorkerNameFormat, moduleJobConfig.getModuleCode(),
			IPUtils.getLocalIP());
		// jobWorker的并发数量
		if (StringUtils.isNotBlank(moduleJobConfig.getJobWorkerSize())) {
			this.jobWorkerSize = Integer.parseInt(moduleJobConfig.getJobWorkerSize());
		}

		// 本地job处理器
		localJobDataProcess = new LocalJobDataProcess(this);
		// 高优先级job处理器
		priorityJobDataProcess = new PriorityJobDataProcess(this);
		// 状态为(F,L)的job处理器，异步执行
		pendingJobDateProcess = new PendingJobDataProcess(this);
		// 状态为(C)的job处理器
		newJobDataProcess = new NewJobDataProcess(this);
	}

	/**
	 * 启动moduleWorker，初始化jobWorker
	 */
	public synchronized void start() {
		// 避免重复启动
		if (this.isRuning) {
			return;
		}
		log.info("moduleWorker moduleName={}, moduleWorkerName={} is starting ... ", this.moduleJobConfig.getModuleCode(),
				this.moduleWorkerName);

		// 创建jobWorkerSize数量的子线程
		this.jobWorkers = new JobWorker[this.jobWorkerSize];
		for (int i = 0; i < this.jobWorkerSize; i++) {
			this.jobWorkers[i] = new JobWorker(this.moduleJobConfig, i, this.moduleWorkerName);
			this.jobWorkers[i].start();
		}
		this.isRuning = true;

		// 启动jobBoss线程，永久运行中，除非调用stop方法停止
		Thread currentThread = new Thread(this);
		currentThread.start();

		log.info("JobBoss moduleName={}, moduleWorkerName={} start successful ", this.moduleJobConfig.getModuleCode(),
				this.moduleWorkerName);
	}

	/**
	 * 核心自启动逻辑
	 */
	@Override
	public void run() {
		while (this.isRuning) {
			// 获取上次执行时间和当前时间相差秒数
			int seconds = YYDateUtils.getIntervalSecond(this.lastExecuteTime, System.currentTimeMillis());
			// 小于30秒则控制下速度
			if (seconds < LOOP_SLEEP_SECONDS) {
				// sleep下
				SecureRandom random = new SecureRandom();
				int sleepSeconds = LOOP_SLEEP_SECONDS - seconds + random.nextInt(10);
				LockSupport.parkNanos(JobConstant.NANO_SECONDS * sleepSeconds);
			}

			log.info("moduleWorker start loop running, moduleCode={}, moduleWorkerName={}, runCount={}",
					this.moduleJobConfig.getModuleCode(), this.moduleWorkerName, ++runCount);

			// 状态为(F,L)的job处理器开始执行,异步处理
			this.pendingJobDateProcess.execute(runCount);
			// 本地job处理器开始执行
			this.localJobDataProcess.execute(runCount);
			// 状态为(C)的job处理器开始执行
			this.newJobDataProcess.execute(runCount);

			// 最近一次执行时间
			this.lastExecuteTime = System.currentTimeMillis();

			log.info("moduleWorker end loop running, projectCode={}, moduleCode={}, moduleWorkerName={}, runCount={}",
				this.projectCode,
				this.moduleJobConfig.getModuleCode(), this.moduleWorkerName, runCount);
		}
	}
	
	/**
	 * 高优先级任务处理器，提交的任务在本机立即执行
	 * 
	 * @param job
	 */
	public void executePriorityJob(JobDomain job) {
		priorityJobDataProcess.execute(job);
	}

	/**
	 * 将job放入JobWorker,等待执行
	 * 
	 * @param job
	 */
	public void assignJobToWoker(JobDomain job) {
		if (job == null) {
			return;
		}
		int tryCount = 1;
		while (this.isRuning) {
			// 获取jobWorker
			JobWorker worker = nextJobWoker();
			if (worker.putJob(job)) {
				log.debug("moduleWorker put job  to JobWorker, jobId={}, jobBoss={}, woker={}", job.getId(),
						this.getModuleWorkerName(), worker.getJobWorkerName());
				break;
			}
			// 如果放队列失败，则稍后重试
			else if (tryCount++ % this.jobWorkerSize == 0) {
				log.warn("moduleWorker put job to JobWorker failed, moduleWorker={}, jobId={}", this.getModuleWorkerName(),
						job.getId());
				// sleep 15 s
				LockSupport.parkNanos(JobConstant.NANO_SECONDS * 15);

//				try {
//					Thread.sleep(30000L);
//				} catch (Exception e) {
//					log.warn("moduleWorker put job to JobWorker sleep failed", e);
//				}
			}
		}
	}

	/**
	 * 获取jobWorker<br>
	 * 简单的取模逻辑获取jobWorker
	 * 
	 * @return
	 */
	private JobWorker nextJobWoker() {
		// 取模获取索引
		int index = this.wokerIndex.incrementAndGet() % this.jobWorkerSize;
		return this.jobWorkers[index];
	}

	public String getModuleWorkerName() {
		return this.moduleWorkerName;
	}

	public Queue<JobDomain> getLocalJobQueue() {
		return localJobQueue;
	}

	public String getProjectCode() {
		return projectCode;
	}

	public ModuleXmlConfig getModuleJobConfig() {
		return moduleJobConfig;
	}
	
}
